package com.dayu.tcp.service;

import com.dayu.tcp.listener.ChannelEventListener;
import com.dayu.tcp.listener.EventBehavior;
import com.dayu.tcp.listener.ExceptionEventListener;
import com.dayu.tcp.listener.MessageEventListener;
import com.dayu.tcp.logging.Log;
import com.dayu.tcp.logging.LogFactory;
import io.netty.channel.ChannelHandlerContext;
import lombok.Getter;
import lombok.NonNull;

import java.util.EventListener;

/**
 * @author yehangtech
 * @date 2023/11/26 13:33
 * @description 事件分发器(分发消息 、 通道 、 异常事件给相应的监听器, 允许使用线程池模型以使worker线程迅速返回)
 */
public class EventDispatcher {

    private static final Log log = LogFactory.getLog(EventDispatcher.class);

    @Getter
    private final Service service;

    public EventDispatcher(@NonNull Service service) {
        this.service = service;
    }

    /**
     * 转发通道事件: 通道连接或关闭
     * @param ctx
     * @param channel
     */
    public void dispatchChannelEvent(final ChannelHandlerContext ctx, final WrappedChannel channel) {
        if (service.isOpenExecutor()) {
            service.channelExecutor.execute(() -> doChannelEvent(ctx, channel));
        } else {
            doChannelEvent(ctx, channel);
        }
    }

    private void doChannelEvent(final ChannelHandlerContext ctx, final WrappedChannel channel) {
        final boolean isConnected = (channel != null && channel.isActive());
        try {
            for (EventListener listener : service.getEventListenerList()) {
                if (listener instanceof ChannelEventListener) {
                    ChannelEventListener channelEventListener = (ChannelEventListener) listener;
                    EventBehavior eventBehavior;
                    if (isConnected) {
                        eventBehavior = channelEventListener.channelActive(ctx, channel);
                    } else {
                        eventBehavior = channelEventListener.channelInActive(ctx, channel);
                    }
                    if (EventBehavior.BREAK.equals(eventBehavior)) {
                        break;
                    }
                }
            }
        } catch (Exception ex) {
            dispatchExceptionCaught(ctx, channel, ex);
        }
    }

    /**
     * 转发异常事件
     *
     * @param ctx
     * @param channel
     * @param cause
     */
    public void dispatchExceptionCaught(final ChannelHandlerContext ctx, final WrappedChannel channel, final Throwable cause) {
        if (service.isOpenExecutor()) {
            service.exceptionExecutor.execute(() -> doExceptionEvent(ctx, channel, cause));
        } else {
            doExceptionEvent(ctx, channel, cause);
        }
    }

    private void doExceptionEvent(final ChannelHandlerContext ctx, final WrappedChannel channel, final Throwable cause) {
        try {
            for (EventListener listener : service.getEventListenerList()) {
                if (listener instanceof ExceptionEventListener) {
                    ExceptionEventListener exceptionEventListener = (ExceptionEventListener) listener;

                    EventBehavior eventBehavior = exceptionEventListener.exceptionCaught(ctx, channel, cause);
                    if (EventBehavior.BREAK.equals(eventBehavior)) {
                        break;
                    }
                }
            }
        } catch (Exception ex) {
            log.error("Failed to dispatch exception event on channel '" + channel.id().asShortText() + "'. ", ex);
        }
    }

    /**
     * 转发消息事件：收到入站消息
     *
     * @param ctx
     * @param channel
     * @param msg
     */
    public void dispatchMessageEvent(final ChannelHandlerContext ctx, final WrappedChannel channel, final Object msg) {
        if (service.isOpenExecutor()) {
            service.exceptionExecutor.execute(() -> doMessageEvent(ctx, channel, msg));
        } else {
            doMessageEvent(ctx, channel, msg);
        }
    }

    private void doMessageEvent(final ChannelHandlerContext ctx, final WrappedChannel channel, final Object msg) {
        try {
            for (EventListener listener : service.getEventListenerList()) {
                if (listener instanceof MessageEventListener) {
                    MessageEventListener messageEventListener = (MessageEventListener) listener;
                    EventBehavior eventBehavior = messageEventListener.channelRead(ctx, channel, msg);
                    if (EventBehavior.BREAK.equals(eventBehavior)) {
                        break;
                    }
                }
            }
        } catch (Exception ex) {
            dispatchExceptionCaught(ctx, channel, ex);
        }
    }

}
